package org.ws.task.schedule.dispatcher;

import com.google.common.collect.Lists;
import com.google.common.hash.Hashing;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.common.KeyUtils;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.TaskConfig;
import org.ws.task.model.TaskItem;

import java.util.List;

public class HashDispatcher extends AbstractDispatcher{



    private static Logger log = LoggerFactory.getLogger(HashDispatcher.class.getName());

    @Override
    public TaskConfig.DispatchStrategy strategy() {
        return TaskConfig.DispatchStrategy.HASH;
    }

    @Override
    protected List<NodeMetadata> getNode(TaskItem item, List<NodeMetadata> taskWorkerNodes) {
        Object data = item.getData();
        String key;
        if(data instanceof String){
            key = String.valueOf(data);
        }
        else {
            key = KeyUtils.getTaskKey(data);
        }

        final int size = taskWorkerNodes.size();
        final int keyHash = Hashing.consistentHash(hash(key),size * 8);
        final int index = keyHash % size;
        log.debug("nodeSize:{},keyHash:{},index:{}",size,keyHash,index);
        return Lists.newArrayList(taskWorkerNodes.get(index));
    }

    private int hash(String key){
        int hashCode = key.hashCode();
        int h = hashCode != Integer.MIN_VALUE ? Math.abs(hashCode) : Integer.MIN_VALUE;
        return h ^ (h >>> 16);
    }
}
